package com.rds.benchmark;

import com.rds.benchmark.config.ClientType;
import com.rds.benchmark.connector.*;
import com.rds.benchmark.operator.*;
import com.rds.benchmark.util.MakeValue;
import com.rds.benchmark.util.Pxx;
import io.lettuce.core.api.StatefulConnection;
import io.lettuce.core.api.StatefulRedisConnection;
import io.lettuce.core.cluster.api.StatefulRedisClusterConnection;
import org.redisson.api.RedissonClient;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisCluster;
import redis.clients.jedis.util.JedisClusterCRC16;

import java.nio.charset.StandardCharsets;
import java.util.Date;
import java.util.Random;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.CyclicBarrier;

public class RdsBenchmark {
    private final static int DEFAULT_THREADS = 50;

    private final static long DEFAULT_MILLI_TIME = 120 * 1000;

    private final static Random RANDOM = new Random();

    private volatile static int Start = 0;
    private volatile static int Stop = 16383;

    private volatile static CyclicBarrier startBarrier;

    private volatile static JedisConnector connector;


    private final static int PXX_MAX = 8;
    private final static Pxx[] PXXES = new Pxx[PXX_MAX];

    static {
        for (int i = 0; i < PXX_MAX; ++i) {
            PXXES[i] = new Pxx(5000, 9000, 9500, 9900);
        }
    }

    private static void init_pxx() {
        for (int i = 0; i < PXX_MAX; ++i) {
            PXXES[i].init();
        }
    }

    private static void Usage() {
        System.out.println("\nUsage: java -jar benchmark-1.x.jar [-h host] [-p port] [-a password] [-t thread] [-c cmd] [--time duration] [--range start-stop]\n");
        System.out.println(" -h or --host   <hostname>                       Server hostname (default 127.0.0.1)");
        System.out.println(" -p or --port   <port>                           Server port (default 6379)");
        System.out.println(" -m or --mode   <connection mode>                Connection mode, optional values: single, sentinel, cluster (default single)"); // TODO 新加
        System.out.println(" -a or --pass   <password>                       The password to login (default null)");
        System.out.println(" -n or --mastername   <master name>              Server mastername (default WebSession)");
        System.out.println(" -s or --sentinelpassword   <sentinel password>  Server sentinelpassword (default null)");
        System.out.println(" -t or --thread   <clients>                      Number of parallel connections (default " + DEFAULT_THREADS + ")");
        System.out.println(" -d <size>                                       Data size of SET/GET value in bytes (default 6)");
        System.out.println(" --time   <duration>                             Test run time, in seconds (default 120)");
        System.out.println(" --range  <start-stop>                           The slot range of the key (default 0-16383)");
        System.out.println(" --client  <clienttype>                          The client type : jedis , redisson , lettuce");
        System.out.println(" --number                                        Total number of requests");
        System.out.println(" -c or --command  <cmd> <cmd-list>               Data type to test, optional values: string, list, set, hash, zset, publish, w1r9\\");
        System.out.println("                                                 , stable(7 * 24 stability test), all(include string,list,set,hash,zset,publish)\\");
        System.out.println("                                                 , (default string)");
        System.out.println(" --help                                          this message");
        System.out.println();
        System.exit(0);
    }

    public static void main(String[] argv) {
        String host = "localhost";
        int port = -1;
        String mode = "single"; //连接模式 single | sentinel | cluster
        int threads = DEFAULT_THREADS;
        long milliTime = DEFAULT_MILLI_TIME;
        int cmd = 0;

        Integer requests = null;

        String masterName = "WebSession";
        String sentinelPassword = null;
        String password = null;

        // 解析参数
        for (int i = 0; i < argv.length; i++) {
            if ("-h".equalsIgnoreCase(argv[i])
                    || "--host".equalsIgnoreCase(argv[i])) {
                if (i + 1 < argv.length) {
                    host = argv[++i];
                } else {
                    System.out.println("invalid parameter -h");
                    Usage();
                }
            } else if ("-p".equalsIgnoreCase(argv[i])
                    || "--port".equalsIgnoreCase(argv[i])) {
                try {
                    port = Integer.parseInt(argv[++i]);
                } catch (Exception e) {
                    System.out.println("invalid parameter -p: " + e.getMessage());
                    Usage();
                }
            } else if ("-m".equalsIgnoreCase(argv[i])
                    || "--mode".equalsIgnoreCase(argv[i])) {
                if (i + 1 < argv.length) {
                    mode = argv[++i];
                } else {
                    System.out.println("invalid parameter -m");
                    Usage();
                }
            } else if ("-n".equalsIgnoreCase(argv[i])
                    || "--mastername".equalsIgnoreCase(argv[i])) {
                if (i + 1 < argv.length) {
                    masterName = argv[++i];
                } else {
                    System.out.println("invalid parameter -n");
                    Usage();
                }
            } else if ("-s".equalsIgnoreCase(argv[i])
                    || "--sentinelpassword".equalsIgnoreCase(argv[i])) {
                if (i + 1 < argv.length) {
                    sentinelPassword = argv[++i];
                } else {
                    System.out.println("invalid parameter -s");
                    Usage();
                }
            } else if ("-t".equalsIgnoreCase(argv[i])
                    || "--thread".equalsIgnoreCase(argv[i])) {
                try {
                    threads = Integer.parseInt(argv[++i]);
                } catch (Exception e) {
                    System.out.println("invalid parameter -t: " + e.getMessage());
                    Usage();
                }
            } else if ("-c".equalsIgnoreCase(argv[i])
                    || "--command".equalsIgnoreCase(argv[i])) {
                String command = argv[++i];
                if ("string".equalsIgnoreCase(command)) {
                    cmd = 0;
                } else if ("list".equalsIgnoreCase(command)) {
                    cmd = 1;
                } else if ("set".equalsIgnoreCase(command)) {
                    cmd = 2;
                } else if ("hash".equalsIgnoreCase(command)) {
                    cmd = 3;
                } else if ("zset".equalsIgnoreCase(command)) {
                    cmd = 4;
                } else if ("publish".equalsIgnoreCase(command)) {
                    cmd = 5;
                } else if ("w1r9".equalsIgnoreCase(command)) {
                    cmd = 6;
                } else if ("stable".equalsIgnoreCase(command)) {
                    cmd = 90;
                } else if ("all".equalsIgnoreCase(command)) {
                    cmd = 100;
                }
            } else if ("--time".equalsIgnoreCase(argv[i])) {
                try {
                    milliTime = Long.parseLong(argv[++i]) * 1000;
                } catch (Exception e) {
                    System.out.println("invalid parameter --time: " + e.getMessage());
                    Usage();
                }
            } else if ("--range".equalsIgnoreCase(argv[i])) {
                String range = argv[++i];
                try {
                    String[] rs = range.split("-");
                    if (rs != null && rs.length > 0) {
                        Start = Integer.parseInt(rs[0]);
                        if (rs.length > 1) {
                            Stop = Integer.parseInt(rs[1]);
                        }
                    }
                } catch (Throwable t) {
                    System.err.println("Error range: " + range);
                }
                System.out.println("Range of slot is [" + Start + " - " + Stop + "]");
            } else if ("-a".equalsIgnoreCase(argv[i]) || "--pass".equalsIgnoreCase(argv[i])) {
                password = argv[++i];
            } else if ("-d".equalsIgnoreCase(argv[i])) {
                MakeValue.make(argv[++i]);
            } else if ("--number".equalsIgnoreCase(argv[i])) {
                String tNumber = argv[++i];
                try {
                    requests = Integer.parseInt(tNumber);
                } catch (Exception e) {
                    System.err.println("Error number: " + tNumber);
                }
            } else if ("--client".equalsIgnoreCase(argv[i])) {
                ClientType.clientType = argv[++i];
            } else if ("--help".equalsIgnoreCase(argv[i])) {
                Usage();
            } else {
                System.err.println("\nUnknown parameter: " + argv[i] + "\n");
                Usage();
            }
        }

        if (port == -1) {
            port = 6379;
        }

        System.out.println();
        System.out.println("EndPoint: " + host + ":" + port);
        System.out.println("Concurrent clients: " + threads);
        System.out.println();

        // 这里参数都初始化完成了
        /**
         * 做一些事情：
         * 1、判断一下参数的正确性
         * 2、根据requests 和 max_threads 计算出所需要的线程
         **/

        if (requests != null) {
            if (requests < threads) {
                // 执行命令数量小于线程数，让线程数 等于 执行命令数量
                threads = requests;
            }
        }

        try {
            startBarrier = new CyclicBarrier(threads + 1); //需要等待一起执行的进程数量（threads测试进程数量， + 1 为主进行）

            if ("jedis".equalsIgnoreCase(ClientType.clientType)) {
                if ("single".equals(mode)) {
                    connector = new SingleJedisConnector(host, port, threads, password);
                } else if ("sentinel".equals(mode)) {
                    connector = new SentinelJedisConnector(host, port, threads, sentinelPassword, masterName, password);
                } else if ("cluster".equals(mode)) {
                    connector = new ClusterJedisConnector(host, port, threads, password);
                } else {
                    throw new Exception("Connection mode error! allowed values: single,sentinel,cluster. " + " mode=" + mode);
                }
            } else if ("redisson".equalsIgnoreCase(ClientType.clientType)) {
                connector = new RedissonConnector(mode, host, port, threads, password, sentinelPassword, masterName);
            } else if ("lettuce".equalsIgnoreCase(ClientType.clientType)) {
                if ("single".equals(mode)) {
                    connector = new SingleLettuceConnector(host, port, threads, password);
                } else if ("sentinel".equals(mode)) {
                    connector = new SentinelLettuceConnector(host, port, threads, sentinelPassword, masterName, password);
                } else if ("cluster".equals(mode)) {
                    connector = new ClusterLettuceConnector(host, port, threads, password);
                } else {
                    throw new Exception("Connection mode error! allowed values: single,sentinel,cluster. " + " mode=" + mode);
                }
            }

            if (!connector.test()) {
                throw new Exception("Connection test failed! for connector: " + connector);
            }

            switch (cmd) {
                case 0:
                    if ("jedis".equalsIgnoreCase(ClientType.clientType))
                        testOperators(connector, threads, milliTime, StringOperator.class, "string", requests);
                    else if ("redisson".equalsIgnoreCase(ClientType.clientType))
                        testOperators(connector, threads, milliTime, StringRedissonOperator.class, "string", requests);
                    else if ("lettuce".equalsIgnoreCase(ClientType.clientType))
                        testOperators(connector, threads, milliTime, StringLettuceOperator.class, "string", requests);
                    break;
                case 1:
                    testOperators(connector, threads, milliTime, ListOperator.class, "list", requests);
                    break;
                case 2:
                    testOperators(connector, threads, milliTime, SetOperator.class, "set", requests);
                    break;
                case 3:
                    testOperators(connector, threads, milliTime, HashOperator.class, "hash", requests);
                    break;
                case 4:
                    testOperators(connector, threads, milliTime, ZsetOperator.class, "zset", requests);
                    break;
                case 5:
                    testOperators(connector, threads, milliTime, PublishOperator.class, "publish", requests);
                    break;
                case 6:
                    testOperators(connector, threads, milliTime, OneWriteNineReadOperator.class, "w1r9", requests);
                    break;
                case 90:
                    if ("jedis".equalsIgnoreCase(ClientType.clientType))
                        stableOperators(connector, threads, milliTime, StableStringOperator.class, "string");
                    else if ("redisson".equalsIgnoreCase(ClientType.clientType))
                        stableOperators(connector, threads, milliTime, StableStringRedissonOperator.class, "string");
                    else if ("lettuce".equalsIgnoreCase(ClientType.clientType))
                        stableOperators(connector, threads, milliTime, StableStringLettuceOperator.class, "string");
                    break;
                case 100:
                    testOperators(connector, threads, milliTime, StringOperator.class, "string", requests);
                    testOperators(connector, threads, milliTime, ListOperator.class, "list", requests);
                    testOperators(connector, threads, milliTime, SetOperator.class, "set", requests);
                    testOperators(connector, threads, milliTime, HashOperator.class, "hash", requests);
                    testOperators(connector, threads, milliTime, ZsetOperator.class, "zset", requests);
                    testOperators(connector, threads, milliTime, PublishOperator.class, "publish", requests);
                    break;
                default:
                    System.out.println("Unknown command");
                    Usage();
            }
        } catch (Throwable t) {
            System.out.println("Error occur: " + t);
            Usage();
        }
    }

    private static String createKey(String prefix) {
        String key = prefix;

        if (Start > 0 || Stop < 16383) {
            int slot = CRC16.getSlot(key.getBytes(StandardCharsets.UTF_8));
            while (slot < Start || slot > Stop) {
                key = prefix + "-" + RANDOM.nextInt(100000);
                slot = CRC16.getSlot(key.getBytes(StandardCharsets.UTF_8));
            }
        }

        //System.out.println("createKey: " + key );
        return key;
    }

    private static void testOperators(JedisConnector connector, final int max_threads, final long milliTime, Class<? extends TypeOperator> operatorClazz, String type, Integer req) throws Exception {

        init_pxx();

        final TypeOperator[] operators = new TypeOperator[max_threads];
        for (int i = 0; i < max_threads; i++) {
            operators[i] = operatorClazz.getDeclaredConstructor().newInstance();
        }

        Thread[] threads = new Thread[max_threads];

        int[] thread_req = new int[max_threads];
        int avg_req = 0;
        if (req != null) {
            // 计算一下每个线程需要执行多少请求
            int srt_req = req % max_threads;
            if (srt_req == 0) {
                avg_req = req / max_threads;
            } else {
                int tmp_avg = req / max_threads;
                for (int i = 0; i < thread_req.length; i++) {
                    if (i <= srt_req) {
                        thread_req[i] = tmp_avg + 1;
                    } else {
                        thread_req[i] = tmp_avg;
                    }
                }
            }
        }

        startBarrier.reset();

        CountDownLatch latch = new CountDownLatch(max_threads);

        // 搞一堆线程出来
        final int final_avg_req = avg_req;
        for (int i = 0; i < max_threads; i++) {
            // final int final_i = i;
            threads[i] = new Thread(Integer.toString(i)) {
                @Override
                public void run() {
                    int idx = Integer.parseInt(Thread.currentThread().getName());
                    int pxx_idx = idx % PXX_MAX;
                    String prefix = UUID.randomUUID().toString();
                    String key = createKey(prefix);
                    TypeOperator operator = operators[idx];
                    Object conn = null;
                    try {
                        conn = connector.getJedis();

                        if (conn instanceof JedisCluster) {
                            conn = new Jedis(((JedisCluster) conn).getConnectionFromSlot(JedisClusterCRC16.getSlot(key)));
                        } else if (conn instanceof RedissonClient) {
                            conn = ((RedissonClient) conn).getBucket(key);
                        } else if (conn instanceof StatefulConnection) {
                            if (conn instanceof StatefulRedisConnection)
                                conn = ((StatefulRedisConnection) conn).sync();
                            else if (conn instanceof StatefulRedisClusterConnection)
                                conn = ((StatefulRedisClusterConnection) conn).sync();
                        }

                        if (req == null)
                            operator.init(conn, key, idx, null);

                        //System.out.println("Thread-" + idx + " barriers waiting count:" + startBarrier.getNumberWaiting());
                        startBarrier.await();

                        //操作前清理之前的数据
                        // operator.clear();

                        // 这里是按照持续时间作为判断条件
                        // 在这里添加基于requests的条件
                        /**
                         * 首先requests和milliTime两个值只能有一个生效，可以默认让milliTime生效。
                         * requests生效，按照当前指定的线程分一下平均每个线程要发送多少个请求。
                         * 1、当requests数量小于max_threads的数量时候，max_threads的数量等于requests
                         * 2、让requests \ max_threads 计算出来的值作为循环条件替换这里即可
                         */
                        if (req == null) {

                            long stop = System.currentTimeMillis() + milliTime;
                            while (stop > System.currentTimeMillis()) {
                                long sub_start = System.nanoTime();

                                operator.operate();

                                sub_start = (System.nanoTime() - sub_start) / 1000;
                                if (sub_start > 0) {
                                    PXXES[pxx_idx].set(sub_start);
                                }
                            }
                        } else {

                            int for_num = final_avg_req;

                            if (for_num == 0) {
                                for_num = thread_req[idx];
                            }
                            for (int j = 0; j < for_num; j++) {
                                long sub_start = System.nanoTime();
                                operator.init(conn, key + idx + j, idx, "req");
                                operator.operate();

                                sub_start = (System.nanoTime() - sub_start) / 1000;
                                if (sub_start > 0) {
                                    PXXES[pxx_idx].set(sub_start);
                                }
                            }
                        }
                    } catch (Throwable t) {
                        System.err.println("Thread-" + idx + " run failed!");
//                        t.printStackTrace();
                    } finally {
                        connector.close(conn);
                        if (req == null)
                            operator.clear(); //清除测试数据
                        operator.endTimes();
                    }
                    latch.countDown();
                }
            };

            threads[i].start();
        }

        Thread state = new Thread(() -> {
            try {
                //System.out.println("State thread barriers waiting count:" + startBarrier.getNumberWaiting());
                startBarrier.await();
            } catch (Throwable t) {
                System.out.println("State thread start failed!");
                t.printStackTrace();
                return;
            }

            System.out.println("Performance " + type + " testing " + (connector.type() != null ? "in " + connector.type() + " model " : "") + "start...");

            long count = 0;
            long size = 0;
            boolean notFirst = false;
            long first = 0;
            long second = 0;
            int cur_temp = req;

            while (latch.getCount() > 0) {
                long begin = System.currentTimeMillis();

                try {
                    Thread.sleep(10000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

                int cur = 0;
                for (int i = 0; i < max_threads; i++) {
                    TypeOperator op = operators[i];
                    int one = op.resetTimes();
                    if (one >= 0) {
                        cur += one;
                    }
                }
                if(cur == 0){
                    cur = cur_temp;
                }else {
                    cur_temp = cur_temp - cur;
                }
                if (latch.getCount() > 0 || req != null) {
                    long duration = (System.currentTimeMillis() - begin);

                    second = first;
                    first = (long) (cur * 1000.0 / duration);
                    System.out.println(new Date() + " -- process " + cur + " times in " + duration + "ms(" + first + "/s)");

                    if (second > 0) {
                        if (!notFirst) {
                            // 去掉第一条记录
                            notFirst = true;
                        } else {
                            // second 为了去掉最后1条记录
                            count += second;
                            ++size;
                        }
                    }
                }
            }

            double p50 = 0;
            double p90 = 0;
            double p95 = 0;
            double p99 = 0;
            long total = 0;
            double duration = 0;
            for (int i = 0; i < PXX_MAX; ++i) {
                long[] data = PXXES[i].get();
                p50 += data[0];
                p90 += data[1];
                p95 += data[2];
                p99 += data[3];
                total += data[4];
                duration += data[5];
            }
            p50 /= PXX_MAX * 1000;
            p90 /= PXX_MAX * 1000;
            p95 /= PXX_MAX * 1000;
            p99 /= PXX_MAX * 1000;
            if (total > 0) {
                duration /= total * 1000;
            } else {
                duration = 0;
            }

            System.out.println(String.format("Performance testing end, average %d/s, latency (AVG: %.3f, P50: %.3f, P90: %.3f, P95: %.3f, P99: %.3f)\n"
                    , (size > 0 ? count / size : 0)
                    , duration
                    , p50, p90, p95, p99
            ));
            Object jedis = connector.getJedis();
            if (jedis instanceof RedissonClient) {
                ((RedissonClient) jedis).shutdown();
            }
        });

        state.start();
        state.join();
    }

    private static long PrintTimestamp = 0;

    private static synchronized boolean needPrint() {
        long now = System.currentTimeMillis() / 1000;
        if (PrintTimestamp != now) {
            PrintTimestamp = now;
            return true;
        }
        return false;
    }

    private static void stableOperators(JedisConnector connector, final int max_threads, final long milliTime, Class<? extends TypeOperator> operatorClazz, String type) throws Exception {

        init_pxx();

        final TypeOperator[] operators = new TypeOperator[max_threads];
        for (int i = 0; i < max_threads; i++) {
            operators[i] = operatorClazz.getDeclaredConstructor().newInstance();
        }

        Thread[] threads = new Thread[max_threads];

        startBarrier.reset();

        CountDownLatch latch = new CountDownLatch(max_threads);

        for (int i = 0; i < max_threads; i++) {
            threads[i] = new Thread(Integer.toString(i)) {
                @Override
                public void run() {
                    int idx = Integer.parseInt(Thread.currentThread().getName());
                    int pxx_idx = idx % PXX_MAX;
                    String prefix = UUID.randomUUID().toString();
                    String key = createKey(prefix);
                    TypeOperator operator = operators[idx];
                    Object conn = null;
                    try {
                        //System.out.println("Thread-" + idx + " barriers waiting count:" + startBarrier.getNumberWaiting());
                        startBarrier.await();

                        boolean first = true;
                        long stop = System.currentTimeMillis() + milliTime;
                        while (stop > System.currentTimeMillis()) {
                            try {
                                conn = connector.getJedis();
                                if (conn instanceof RedissonClient) {
                                    conn = ((RedissonClient) conn).getBucket(key);
                                } else if (conn instanceof StatefulConnection) {
                                    if (conn instanceof StatefulRedisConnection)
                                        conn = ((StatefulRedisConnection) conn).sync();
                                    else if (conn instanceof StatefulRedisClusterConnection)
                                        conn = ((StatefulRedisClusterConnection) conn).sync();
                                }
                                operator.init(conn, key, idx, null);
                                if (first) {
                                    first = false;
                                    //操作前清理之前的数据
                                    operator.clear();
                                }

                                long sub_start = System.nanoTime();

                                operator.operate();

                                sub_start = (System.nanoTime() - sub_start) / 1000;
                                if (sub_start > 0) {
                                    PXXES[pxx_idx].set(sub_start);
                                }

                                if (needPrint()) {
                                    System.out.println(new Date() + " ok.");
                                }
                            } catch (Throwable t) {
                                if (needPrint()) {
                                    System.out.println(new Date() + " Error: " + t.getMessage());
                                }
                            } finally {
                                connector.close(conn);
                            }
                        }
                    } catch (Throwable t) {
                        System.err.println("Thread-" + idx + " run failed!");
//                        t.printStackTrace();
                    } finally {
                        operator.clear(); //清除测试数据
                        operator.endTimes();
                    }
                    latch.countDown();
                }
            };

            threads[i].start();
        }

        Thread state = new Thread(() -> {
            try {
                //System.out.println("State thread barriers waiting count:" + startBarrier.getNumberWaiting());
                startBarrier.await();
            } catch (Throwable t) {
                System.out.println("State thread start failed!");
                t.printStackTrace();
                return;
            }

            System.out.println("Stable " + type + " testing " + (connector.type() != null ? "in " + connector.type() + " model " : "") + "start...");

            long count = 0;
            long size = 0;
            boolean notFirst = false;
            long first = 0;
            long second = 0;

            while (latch.getCount() > 0) {
                long begin = System.currentTimeMillis();

                try {
                    Thread.sleep(10000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

                int cur = 0;
                for (int i = 0; i < max_threads; i++) {
                    TypeOperator op = operators[i];
                    int one = op.resetTimes();
                    if (one >= 0) {
                        cur += one;
                    }
                }
                if (latch.getCount() > 0) {
                    long duration = (System.currentTimeMillis() - begin);

                    second = first;
                    first = (long) (cur * 1000.0 / duration);
//                    System.out.println(new Date() + " -- process " + cur + " times in " + duration + "ms(" + first + "/s)");

                    if (second > 0) {
                        if (!notFirst) {
                            // 去掉第一条记录
                            notFirst = true;
                        } else {
                            // second 为了去掉最后1条记录
                            count += second;
                            ++size;
                        }
                    }
                }
            }

            double p50 = 0;
            double p90 = 0;
            double p95 = 0;
            double p99 = 0;
            long total = 0;
            double duration = 0;
            for (int i = 0; i < PXX_MAX; ++i) {
                long[] data = PXXES[i].get();
                p50 += data[0];
                p90 += data[1];
                p95 += data[2];
                p99 += data[3];
                total += data[4];
                duration += data[5];
            }
            p50 /= PXX_MAX * 1000;
            p90 /= PXX_MAX * 1000;
            p95 /= PXX_MAX * 1000;
            p99 /= PXX_MAX * 1000;
            if (total > 0) {
                duration /= total * 1000;
            } else {
                duration = 0;
            }

            System.out.println(String.format("Stable testing end, average %d/s, latency (AVG: %.3f, P50: %.3f, P90: %.3f, P95: %.3f, P99: %.3f)\n"
                    , (size > 0 ? count / size : 0)
                    , duration
                    , p50, p90, p95, p99
            ));
            Object jedis = connector.getJedis();
            if (jedis instanceof RedissonClient) {
                ((RedissonClient) jedis).shutdown();
            }
        });

        state.start();
        state.join();
    }

}
